package com.raylu.util

import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.Jedis

import java.util


/**
 *
 * Description:
 *
 * Create by lucienoz on 2021/11/26.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
object OffsetManagerUtil {

		def getOffset(topic:String,groupId:String): Map[TopicPartition, Long] ={
			val jedis : Jedis = RedisUtil.getJedisFromPool ()
			jedis.select(PropertiesUtil.load("config.properties").getProperty("redis.offset.db").toInt)
			import scala.collection.JavaConverters._
			val map : Map[TopicPartition, Long] = jedis.hgetAll ( s"offset:${topic}:${groupId}" ).asScala.map {
				case (part, offset) => {
					(new TopicPartition ( topic, part.toInt ), offset.toLong)
				}
			}.toMap
			jedis.close()
			map
		}

		def saveOffset(topic:String,groupId:String,offsetRanges: Array[OffsetRange]): Unit ={
			val jedis : Jedis = RedisUtil.getJedisFromPool ()
			jedis.select(PropertiesUtil.load("config.properties").getProperty("redis.offset.db").toInt)
			val offsetMap : util.HashMap[String, String] = new util.HashMap[String, String]()
			for (offset <- offsetRanges) {
				offsetMap.put(offset.partition.toString,offset.untilOffset.toString)
			}
			jedis.hset(s"offset:${topic}:${groupId}",offsetMap)
			jedis.close()
		}



}
